package net.bwie.vehicle.dws.function1;

import net.bwie.vehicle.dws.bean1.VehicleData;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import java.math.BigDecimal;

// 电池健康度聚合器 - 计算平均电池健康度
/**
 * 电池电量聚合函数类，用于计算车辆数据的平均电池电量
 * 实现了AggregateFunction接口，用于聚合操作
 */
public class BatteryLevelAggregatorFunction implements
        AggregateFunction<VehicleData, Tuple2<BigDecimal, Integer>, Double> {

    /**
     * 创建初始聚合器
     * @return 返回一个初始聚合器，包含电量总和为0和计数为0的Tuple2对象
     */
    @Override
    public Tuple2<BigDecimal, Integer> createAccumulator() {
        return Tuple2.of(BigDecimal.ZERO, 0);
    }

    /**
     * 更新聚合器
     * 将车辆数据中的电池电量加到聚合器的总和中，并增加计数
     * @param value 当前处理的车辆数据对象
     * @param accumulator 当前的聚合器对象
     * @return 返回更新后的聚合器对象
     */
    @Override
    public Tuple2<BigDecimal, Integer> add(VehicleData value, Tuple2<BigDecimal, Integer> accumulator) {
        return Tuple2.of(accumulator.f0.add(BigDecimal.valueOf(value.getBatteryLevel())), accumulator.f1 + 1);
    }

    /**
     * 获取最终结果
     * 计算并返回平均电池电量如果计数为0，则返回0.0
     * @param accumulator 当前的聚合器对象
     * @return 返回平均电池电量
     */
    @Override
    public Double getResult(Tuple2<BigDecimal, Integer> accumulator) {
        return accumulator.f1 == 0 ? 0.0 : accumulator.f0.divide(BigDecimal.valueOf(accumulator.f1), 2, BigDecimal.ROUND_HALF_UP).doubleValue();
    }

    /**
     * 合并两个聚合器
     * 将两个聚合器的电量总和相加，并将计数相加
     * @param a 第一个聚合器对象
     * @param b 第二个聚合器对象
     * @return 返回合并后的聚合器对象
     */
    @Override
    public Tuple2<BigDecimal, Integer> merge(Tuple2<BigDecimal, Integer> a, Tuple2<BigDecimal, Integer> b) {
        return Tuple2.of(a.f0.add(b.f0), a.f1 + b.f1);
    }
}
